package com.dtwave.lnstreaming.sink;

import com.dtwave.lnstreaming.bean.RtDkhtxxb;
import com.dtwave.lnstreaming.connection.ElasticConnector;
import com.dtwave.lnstreaming.utils.ElasticUtils;
import com.dtwave.param.EsparamObj;
import lombok.SneakyThrows;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestHighLevelClient;

import java.io.Serializable;

public class EsSinkFunctionContractInfo implements ElasticsearchSinkFunction<RtDkhtxxb>, Serializable {

    RestHighLevelClient client;

    private EsparamObj esParamObj;
    public EsSinkFunctionContractInfo(EsparamObj esParamObj){
        this.esParamObj = esParamObj;
    }

    @Override
    public void open() throws Exception {
        client = ElasticConnector.getClient(esParamObj.getEsUrl());
    }

    @SneakyThrows
    @Override
    public void process(RtDkhtxxb rtDkhtxxb, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
        IndexRequest indexRequest = ElasticUtils.createIndex(client, rtDkhtxxb);
        if(indexRequest != null){
            requestIndexer.add(indexRequest);
        }
    }

    @Override
    public void close() throws Exception {
           client.close();
    }
}
